package com.acmedcare.framework.exchange.center.consumer;

import com.acmedcare.framework.exchange.center.common.ConvertUtils;
import com.acmedcare.framework.exchange.center.core.Const;
import com.acmedcare.framework.exchange.center.core.PluginContainerMBean;
import com.acmedcare.framework.exchange.center.core.handler.ConsumerHandlerMBean;
import com.acmedcare.framework.exchange.center.core.handler.IAgentPublisher;
import com.acmedcare.framework.exchange.center.entity.bean.Consumer;
import com.acmedcare.framework.exchange.center.entity.bean.PluginStatusEnum;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class ConsumerProxyContainer implements IConsumerProxyContainer {

  Boolean loaded = false;
  final RabbitTemplate rabbitTemplate;
  final ConcurrentHashMap<String, ConsumerProxyMetadata> consumerHashMap;
  final PluginContainerMBean pluginContainer;
  final ObjectMapper objectMapper;
  final IAgentPublisher agentPublisher;
  final ConsumerHandlerMBean consumerHandlerMBean;

  public ConsumerProxyContainer(RabbitTemplate rabbitTemplate,
      PluginContainerMBean pluginContainer, ObjectMapper objectMapper,
      IAgentPublisher agentPublisher,
      ConsumerHandlerMBean consumerHandlerMBean) {

    this.rabbitTemplate = rabbitTemplate;
    this.pluginContainer = pluginContainer;
    this.objectMapper = objectMapper;
    this.agentPublisher = agentPublisher;
    this.consumerHandlerMBean = consumerHandlerMBean;
    consumerHashMap = new ConcurrentHashMap<>();
  }

  @Override
  public void init(List<Consumer> consumers) {

    if (loaded) {

      return;
    }

    synchronized (loaded) {

      if (loaded) {

        return;
      }

      for (Consumer consumer : consumers) {

        //初始化队列，并且绑定交换器
        if (!consumer.getGroupName().equals(System.getenv(Const.GroupName))) {
          log.info("接受消费器指定环境与当前环境不匹配，消费器指令环境为:" + consumer.getGroupName() + ",当前环境为:" + System
              .getenv(Const.GroupName));
          continue;
        }

        createProxy(consumer);
      }

      loaded = true;
    }
  }

  @Override
  public void unload(Consumer consumer) {

    if (consumer == null) {

      return;
    }

    if (!loaded) {

      log.warn("消费器尚未初始化，无法完成卸载");
      return;
    }

    ConsumerProxyMetadata proxyMetaData = getConsumer(consumer.getName());
    if (proxyMetaData == null) {

      log.warn(String.format("消费器：%s,已经完成卸载，无需再次卸载", consumer.getName()));
      return;
    }

    synchronized (proxyMetaData) {

      if (proxyMetaData.getStatus().equals(PluginStatusEnum.UNLOAD)) {

        consumerHashMap.remove(consumer.getName().toLowerCase());
        return;
      }

      proxyMetaData.getListenerContainer().stop();
      proxyMetaData.setStatus(PluginStatusEnum.UNLOAD);
      consumerHashMap.remove(consumer.getName().toLowerCase());
    }

    log.info(String.format("消费器：%s,卸载成功", consumer.getName()));
  }

  @Override
  public synchronized void load(Consumer consumer) {

    if (!loaded) {

      List<Consumer> consumers = new ArrayList<>();
      consumers.add(consumer);
      init(consumers);
      return;
    }

    ConsumerProxyMetadata proxyMetaData = getConsumer(consumer.getName());
    if (proxyMetaData == null) {

      createProxy(consumer);
      return;
    }

    //防止加载时，产生了同一对象的卸载，则会产生代理被卸载，但是检查出结果还是加载的假象，从而导致插件不加载
    if (proxyMetaData.getStatus().equals(PluginStatusEnum.LOADED)
        && proxyMetaData.getVersion().equals(consumer.getVersion())) {

      log.warn(
          String.format("消费器：%s,版本：%s已经完成加载，无需再次加载", consumer.getName(), consumer.getVersion()));
      return;
    }

    if (!proxyMetaData.getStatus().equals(PluginStatusEnum.UNLOAD)) {

      unload(consumer);
    }

    createProxy(consumer);
  }

  private synchronized void createProxy(Consumer consumer) {

    ConsumerProxyMetadata metadata = getConsumer(consumer.getName());
    if (metadata != null) {

      return;
    }

    String currentEnv = System.getenv(Const.GroupName);
    if (!consumer.getGroupName().equals(currentEnv)) {

      log.info(String.format("当前环境:%s与消费器所处的环境:%s不一致，固拒绝加载消费器", currentEnv));
      return;
    }

    ConsumerProxyMetadata proxyMetaData;
    proxyMetaData = ConvertUtils.convertType(consumer, ConsumerProxyMetadata.class);
    declareQueue(proxyMetaData);
    declareListener(proxyMetaData);
    proxyMetaData.setStatus(PluginStatusEnum.LOADED);
    consumerHashMap.put(proxyMetaData.getName().toLowerCase(), proxyMetaData);
    log.info(String.format("消费器：%s,加载完成！", proxyMetaData.getName()));
  }

  @Override
  public ConsumerProxyMetadata getConsumer(String consumerName) {

    return consumerHashMap.get(consumerName.toLowerCase());
  }

  private void declareQueue(ConsumerProxyMetadata consumer) {

    RabbitAdmin admin = new RabbitAdmin(rabbitTemplate.getConnectionFactory());
    FanoutExchange exchange = new FanoutExchange(Const.TaskExchangePrefix + consumer.getTaskName());
    admin.declareExchange(exchange);

    Queue queue = new Queue(getConsumerName(consumer.getName()));
    admin.declareQueue(queue);
    Binding binding = BindingBuilder.bind(queue).to(exchange);
    admin.declareBinding(binding);
  }

  private void declareListener(ConsumerProxyMetadata proxyMetadata) {

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(rabbitTemplate.getConnectionFactory());
    container.setQueueNames(getConsumerName(proxyMetadata.getName()));

    Consumer consumer = ConvertUtils.convertType(proxyMetadata, Consumer.class);
    MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(
        new ConsumerMessageListener(consumer, consumerHandlerMBean));
    listenerAdapter.setDefaultListenerMethod("invoke");
    container.setMessageListener(listenerAdapter);
    container.start();
    proxyMetadata.setListenerContainer(container);
  }

  private String getConsumerName(String name) {

    return String.format("%s.%s.%s", Const.ConsumerExchangeName, name,
        System.getenv(Const.GroupName));
  }
}
